package com.bujunjie.study.realtime.dws.app;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.NumberUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.bujunjie.study.realtime.common.base.BaseApp;
import com.bujunjie.study.realtime.common.constant.FlinkConstant;
import com.bujunjie.study.realtime.common.function.Object2JsonStringFunction;
import com.bujunjie.study.realtime.common.util.FlinkSinkUtil;
import com.bujunjie.study.realtime.dws.app.model.sink.TradeSkuOrderModel;
import com.bujunjie.study.realtime.dws.app.model.vo.TradeOrderDetailVO;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Date;

/**
 * <p>sku粒度下单业务过程聚合统计</p>
 *
 * @author bu.junjie
 * @version 1.0.0
 * @createTime 2025/9/19 8:48
 */
public class DwsTradeSkuOrderWindow extends BaseApp {

    public static void main(String[] args) throws Exception {
        new DwsTradeSkuOrderWindow().start(10029, 4, "dws_trade_sku_order_window", FlinkConstant.TOPIC_DWD_TRADE_ORDER_DETAIL);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaStrDS) {

        //  过滤后的流数据
        SingleOutputStreamOperator<TradeOrderDetailVO> filterDS = kafkaStrDS.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String value) throws Exception {
                return StrUtil.isNotBlank(value);
            }
        }).map(new MapFunction<String, TradeOrderDetailVO>() {
            @Override
            public TradeOrderDetailVO map(String value) throws Exception {
                return JSON.parseObject(value, new TypeReference<TradeOrderDetailVO>() {
                });
            }
        });

        KeyedStream<TradeOrderDetailVO, String> orderIdKeyDS = filterDS.keyBy(TradeOrderDetailVO::getId);

        //  数据流去重
        SingleOutputStreamOperator<TradeOrderDetailVO> distinctDS = orderIdKeyDS.process(new KeyedProcessFunction<String, TradeOrderDetailVO, TradeOrderDetailVO>() {

            private ValueState<TradeOrderDetailVO> valueState;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<TradeOrderDetailVO> valueStateDescriptor = new ValueStateDescriptor<>("valueState", TradeOrderDetailVO.class);
                valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.seconds(10)).build());
                valueState = getRuntimeContext().getState(valueStateDescriptor);
            }

            @Override
            public void processElement(TradeOrderDetailVO data, KeyedProcessFunction<String, TradeOrderDetailVO, TradeOrderDetailVO>.Context ctx, Collector<TradeOrderDetailVO> out) throws Exception {
                TradeOrderDetailVO value = valueState.value();
                if (ObjectUtil.isNotNull(value)) {
                    //  代表当前订单数据重复,将里面关于金额数据进行取反
                    value.setSplitActivityAmount(StrUtil.format("-{}", value.getSplitActivityAmount()));
                    value.setSplitCouponAmount(StrUtil.format("-{}", value.getSplitCouponAmount()));
                    value.setSplitPaymentAmount(StrUtil.format("-{}", value.getSplitPaymentAmount()));
                    value.setSplitOriginalAmount(StrUtil.format("-{}", value.getSplitOriginalAmount()));
                    out.collect(value);
                }
                out.collect(data);
                valueState.update(data);
            }
        });

        // 增加流水印
        SingleOutputStreamOperator<TradeOrderDetailVO> watermarksDS = distinctDS.assignTimestampsAndWatermarks(WatermarkStrategy.<TradeOrderDetailVO>forBoundedOutOfOrderness(Duration.ofSeconds(2 * 60)).withTimestampAssigner(new SerializableTimestampAssigner<TradeOrderDetailVO>() {
            @Override
            public long extractTimestamp(TradeOrderDetailVO orderDetailVO, long recordTimestamp) {
                return orderDetailVO.getTs() * 1000;
            }
        }));

        SingleOutputStreamOperator<TradeSkuOrderModel> mapDS = watermarksDS.map(new MapFunction<TradeOrderDetailVO, TradeSkuOrderModel>() {
            @Override
            public TradeSkuOrderModel map(TradeOrderDetailVO value) throws Exception {
                TradeSkuOrderModel tradeSkuOrderModel = BeanUtil.copyProperties(value, TradeSkuOrderModel.class);
                tradeSkuOrderModel.setOrderAmount(NumberUtil.toBigDecimal(value.getSplitTotalAmount()));
                return tradeSkuOrderModel;
            }
        });

        // 流 id 集合
        KeyedStream<TradeSkuOrderModel, String> skuIdKeyDS = mapDS.keyBy(TradeSkuOrderModel::getSkuId);

        WindowedStream<TradeSkuOrderModel, String, TimeWindow> windowDS = skuIdKeyDS.window(TumblingProcessingTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)));

        SingleOutputStreamOperator<TradeSkuOrderModel> reduceDS = windowDS.reduce(new ReduceFunction<TradeSkuOrderModel>() {
            @Override
            public TradeSkuOrderModel reduce(TradeSkuOrderModel value1, TradeSkuOrderModel value2) throws Exception {
                value1.setOriginalAmount(value1.getOriginalAmount().add(value2.getOriginalAmount()));
                value1.setActivityReduceAmount(value1.getActivityReduceAmount().add(value2.getActivityReduceAmount()));
                value1.setCouponReduceAmount(value1.getCouponReduceAmount().add(value2.getCouponReduceAmount()));
                value1.setOrderAmount(value1.getOrderAmount().add(value2.getOrderAmount()));
                return value1;

            }
        }, new WindowFunction<TradeSkuOrderModel, TradeSkuOrderModel, String, TimeWindow>() {
            @Override
            public void apply(String key, TimeWindow window, Iterable<TradeSkuOrderModel> elements, Collector<TradeSkuOrderModel> out) throws Exception {
                TradeSkuOrderModel orderBean = elements.iterator().next();
                orderBean.setStartTime(DateUtil.format(new Date(window.getStart()), DatePattern.NORM_DATETIME_PATTERN));
                orderBean.setEndTime(DateUtil.format(new Date(window.getEnd()), DatePattern.NORM_DATETIME_PATTERN));
                orderBean.setCurrentDate(DateUtil.format(new Date(window.getEnd()), DatePattern.NORM_DATE_PATTERN));
                out.collect(orderBean);
            }
        });


        reduceDS
                .map(new Object2JsonStringFunction<>())
                .sinkTo(FlinkSinkUtil.getDorisSink("dws_trade_province_order_window"));



    }
}
